package com.mate.cloud.redis;


import com.alibaba.fastjson2.JSON;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import com.mate.cloud.core.util.CollectionUtils;
import com.mate.cloud.redis.constants.BloomFilterCache;
import com.mate.cloud.redis.constants.RedisKeyConstants;
import com.mate.cloud.redis.dal.model.City;
import com.mate.cloud.redis.service.ICityService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;


@Slf4j
@Component
public class LoadDataApplicationRunner implements ApplicationRunner {

    @Autowired
    ICityService cityService;

    @Autowired
    StringRedisTemplate redisTemplate;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("=========begin load city data to Redis===========");
        List<City> cityList = cityService.list();

        cacheCityList(RedisKeyConstants.CITY_KEY, cityList);

        // 布隆过滤器
        BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(StandardCharsets.UTF_8), 10000000, 0.03);

        cityList.parallelStream().forEach(city -> {
            bloomFilter.put(String.format(RedisKeyConstants.CITY_KEY, city.getId()));
        });
        BloomFilterCache.cityBloom = bloomFilter;
        log.info("=========finish load city data to Redis===========");
    }

    /**
     * 数据缓存
     *
     * @param cityList
     */
    private void cacheCityList(String redisKey, List<City> cityList) {

        // 1. 空值校验（提前返回，避免无效操作）
        if (CollectionUtils.isEmpty(cityList) || !redisKey.contains("%s")) {
            log.warn("城市列表为空，或 Key 模板缺少 %s 占位符，不执行缓存", redisKey);
            return;
        }

        // 2. 步骤1：生成所有需要操作的 Key（城市专属 Key）
        List<String> cityKeys = cityList.stream().map(city -> String.format(redisKey, city.getId())).collect(Collectors.toList());
        log.info("待缓存城市 Key 列表：{}", cityKeys);

        // 3. 步骤2：批量删除旧 Key（避免新旧数据共存）
        // 用 delete(Collection) 批量删除，比循环单个删除效率高
        Long deletedCount = redisTemplate.delete(cityKeys);
        log.info("批量删除旧城市缓存，成功删除 Key 数量：{}", deletedCount);

        // 4. 步骤3：pipeline 批量写入新数据（核心优化：移除手动 open/closePipeline）
        log.info("开始初始化城市到 Redis，待缓存城市数量：{}", cityList.size());
        redisTemplate.executePipelined((RedisCallback<Void>) redisConnection -> {
            // 无需手动 openPipeline()：executePipelined 内部已处理
            for (City city : cityList) {
                // 生成当前城市的 Key
                byte[] keyBytes = String.format(redisKey, city.getId()).getBytes(StandardCharsets.UTF_8);
                // 城市对象转 JSON 字节数组（与读取时编码一致）
                byte[] valueBytes = JSON.toJSONString(city).getBytes(StandardCharsets.UTF_8);
                // lPush：写入 List 类型（每个 Key 对应一个 List，存 1 个元素）
                redisConnection.lPush(keyBytes, valueBytes);
            }
            // 无需手动 closePipeline()：executePipelined 内部已处理
            return null; // 写入操作无需返回值，泛型声明为 Void
        });

        log.info("城市缓存初始化完成，Key 模板：{}，缓存城市数量：{}", redisKey, cityList.size());
        log.info("############################################################");
    }


}
